package ar.edu.itba.pod.legajo46470.balance;

import static com.google.common.base.Preconditions.checkNotNull;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;

import org.apache.log4j.Logger;

import ar.edu.itba.balance.api.AgentsTransfer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.event.EventInformation;
import ar.edu.itba.event.RemoteEventDispatcher;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.agent.runner.Agent;
import ar.edu.itba.pod.legajo46470.cluster.node.ClusterNode;
import ar.edu.itba.pod.legajo46470.communication.ClusterCommunication;
import ar.edu.itba.pod.legajo46470.event.DistributedEventDispatcher;
import ar.edu.itba.pod.legajo46470.simulation.DistributedSimulation;

public class AgentsTransferImpl extends UnicastRemoteObject implements AgentsTransfer {
	private static final long serialVersionUID = -6344457335961615223L;
	private final static Logger LOGGER = Logger.getLogger(AgentsTransferImpl.class);
	private ClusterNode clusterNode;
	private ClusterCommunication<AgentsTransfer> clusterComm;
	private ClusterCommunication<RemoteEventDispatcher> clusterCommEvents;
	
	public AgentsTransferImpl(ClusterNode clusterNode) throws RemoteException {
		super();
		checkNotNull(clusterNode, "Cluster Node cannot be null");
		this.clusterNode = clusterNode;
		this.clusterComm = new ClusterCommunication<AgentsTransfer>(clusterNode, Node.AGENTS_TRANSFER);
		this.clusterCommEvents = new ClusterCommunication<RemoteEventDispatcher>(clusterNode, Node.DISTRIBUTED_EVENT_DISPATCHER);
	}
	@Override
	public void runAgentsOnNode(List<NodeAgent> agents) throws RemoteException {
		
		final NodeInformation sourceNode = agents.get(0).node();
		final List<NodeAgent> sourceAgents = agents;
		
		DistributedSimulation ds = (DistributedSimulation) clusterNode.getSimulation();
		
		LOGGER.info("The following amount of agents were received to run on my node: " + agents.size());

		// perform sync only if incoming agent is not from self or is a new agent never run in any node
		if (sourceNode != null && !sourceNode.equals(clusterNode.getNodeInfo())) {
				BlockingQueue<Object> agentQueue;
				RemoteEventDispatcher remoteDispatcher;
				Set<EventInformation> syncEventsFrom, syncEventsTo;
				syncEventsFrom = syncEventsTo = null;
				remoteDispatcher = (RemoteEventDispatcher) clusterCommEvents.getStub(sourceNode);
				try {
					// events that owner of incoming agents may have processed while sync operation was on
					syncEventsFrom = remoteDispatcher.newEventsFor(clusterNode.getNodeInfo());
					// events that must be flushed to owner of incoming agents due to sync operation stopping broadcast of events
					syncEventsTo = clusterNode.getRemoteEventDispatcher().newEventsFor(sourceNode);
				} catch (RemoteException e1) {
					LOGGER.warn("Node down: " + sourceNode + ", sync of events incomplete");
					clusterComm.disconnectNode(sourceNode);
				}
				
				// sets the queue for each received agent into my local dispatcher 
				// before publishing syncEventsFrom into my dispatcher 
				for (NodeAgent nodeAgent : sourceAgents ) {
					try {
						agentQueue = remoteDispatcher.moveQueueFor(nodeAgent.agent());
						((DistributedEventDispatcher)(clusterNode.getLocalEventDispatcher())).setAgentQueue(nodeAgent.agent(), agentQueue);
					} catch (RemoteException e1) {
						LOGGER.error("Node down " + nodeAgent.node() + " unable to get agent queue");
						clusterComm.disconnectNode(nodeAgent.node());
					}
				}

				// flush events to node that transfered agents to me
				for (EventInformation e : syncEventsTo) {
					try {
						remoteDispatcher.publish(e);
					} catch (InterruptedException e1) {
						LOGGER.warn("interrupted while sync of events, sync incomplete!");
					} catch (RemoteException e1) {
						LOGGER.warn("Node down: " + sourceAgents.get(0).node());
						clusterComm.disconnectNode(sourceAgents.get(0).node());
					}
				}
				
				// publish to self all events incoming from owner of transfered events
				for (EventInformation e : syncEventsFrom) {
					try {
						clusterNode.getRemoteEventDispatcher().publish(e);
					} catch (InterruptedException e1) {
						LOGGER.warn("Interruped while publishing events for myself during sync");
					} catch (RemoteException e2) {
						// never happens, local call
					}
				}
			}
			
		
		
		// add agents to myself and probably, start them (agentThread)
		ds.addInternal(agents);
	}

	@Override
	public int getNumberOfAgents() throws RemoteException {
		return clusterNode.getSimulation().agentsRunning();
	}

	@Override
	public List<NodeAgent> stopAndGet(int numberOfAgents)
			throws RemoteException {
		LOGGER.info("Coordinator command me to stop: " + numberOfAgents + " agents");
		List<Agent> myAgents = clusterNode.getSimulation().getAgentsRunning();
		List<NodeAgent> extract = new ArrayList<NodeAgent>();
		
		// get numberOfAgents from local simulation, give them to coordinator
		// and remove them from local sim
		for (int i = 0 ; i < numberOfAgents ; i++) {
			extract.add(new NodeAgent(clusterNode.getNodeInfo(), myAgents.get(i)));
			clusterNode.getSimulation().remove(myAgents.get(i));
		}
		
		return extract;
	}

}
